#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <event2/event.h>
#include <event2/listener.h>
#include <event2/bufferevent.h>
#include <event2/buffer.h>
#include <arpa/inet.h>

#include  "sockets/dv_qkd_tcp_server.h"

#include <iostream>
#include <string>
#include <cstdlib>
#include <librdkafka/rdkafkacpp.h>

// 生产者示例
void kafka_producer(const std::string& brokers, const std::string& topic) {
    std::string errstr;
    RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    RdKafka::Conf* tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);

    // 设置 broker
    if (conf->set("bootstrap.servers", brokers, errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << errstr << std::endl;
        exit(1);
    }

    // 创建生产者实例
    RdKafka::Producer* producer = RdKafka::Producer::create(conf, errstr);
    if (!producer) {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        exit(1);
    }

    // 发送消息
    std::string message = "Hello, Kafka!";
    RdKafka::ErrorCode err = producer->produce(
        topic,
        RdKafka::Topic::PARTITION_UA,
        RdKafka::Producer::RK_MSG_COPY,
        const_cast<char*>(message.c_str()),
        message.size(),
        nullptr,
        0,
        0,
        nullptr,
        nullptr
    );

    if (err != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Failed to produce to topic " << topic << ": " <<
            RdKafka::err2str(err) << std::endl;
    } else {
        std::cout << "Enqueued message (" << message.size() << " bytes) for topic " << topic << std::endl;
    }

    // 刷新队列
    producer->flush(5 * 1000); // 等待5秒

    delete producer;
    delete conf;
    delete tconf;
}

// 消费者示例
void kafka_consumer(const std::string& brokers, const std::string& topic) {
    std::string errstr;
    RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    RdKafka::Conf* tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);

    // 设置 broker 和 group.id
    if (conf->set("bootstrap.servers", brokers, errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << errstr << std::endl;
        exit(1);
    }

    if (conf->set("group.id", "test_group", errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << errstr << std::endl;
        exit(1);
    }

    // 创建消费者实例
    RdKafka::Consumer* consumer = RdKafka::Consumer::create(conf, errstr);
    if (!consumer) {
        std::cerr << "Failed to create consumer: " << errstr << std::endl;
        exit(1);
    }

    // 创建 topic 对象
    RdKafka::Topic* rkt = RdKafka::Topic::create(consumer, topic, tconf, errstr);
    if (!rkt) {
        std::cerr << "Failed to create topic: " << errstr << std::endl;
        exit(1);
    }

    // 开始消费
    RdKafka::ErrorCode resp = consumer->start(rkt, 0, RdKafka::Topic::OFFSET_BEGINNING);
    if (resp != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Failed to start consumer: " << RdKafka::err2str(resp) << std::endl;
        exit(1);
    }

    // 消费消息
    for (int i = 0; i < 10; i++) {
        RdKafka::Message* msg = consumer->consume(rkt, 0, 1000);
        switch (msg->err()) {
            case RdKafka::ERR__TIMED_OUT:
                break;
            case RdKafka::ERR_NO_ERROR:
                std::cout << "Read msg at offset " << msg->offset() << std::endl;
                if (msg->key()) {
                    std::cout << "Key: " << *msg->key() << std::endl;
                }
                std::cout << "Payload: " << static_cast<const char*>(msg->payload()) << std::endl;
                break;
            case RdKafka::ERR__PARTITION_EOF:
                std::cout << "Reached end of topic" << std::endl;
                break;
            case RdKafka::ERR__UNKNOWN_TOPIC:
            case RdKafka::ERR__UNKNOWN_PARTITION:
                std::cerr << "Consume failed: " << msg->errstr() << std::endl;
                exit(1);
            default:
                // 处理其他错误
                std::cerr << "Consume failed: " << msg->errstr() << std::endl;
                break;
        }
        delete msg;
    }

    // 停止消费并清理资源
    consumer->stop(rkt, 0);
    delete rkt;
    delete consumer;
    delete conf;
    delete tconf;
}

// 前置声明所有回调函数
static void read_cb_9999(struct bufferevent *bev, void *ctx);
static void read_cb_9998(struct bufferevent *bev, void *ctx);
static void event_cb_9999(struct bufferevent *bev, short events, void *ctx);
static void event_cb_9998(struct bufferevent *bev, short events, void *ctx);
static void accept_conn_cb_9999(struct evconnlistener *listener, evutil_socket_t fd,
                              struct sockaddr *address, int socklen, void *ctx);
static void accept_conn_cb_9998(struct evconnlistener *listener, evutil_socket_t fd,
                              struct sockaddr *address, int socklen, void *ctx);
static void accept_error_cb(struct evconnlistener *listener, void *ctx);

// 端口9999的事件回调
static void event_cb_9999(struct bufferevent *bev, short events, void *ctx) {
    if (events & BEV_EVENT_EOF) {
        printf("[Port 9999] Connection closed\n");
    } else if (events & BEV_EVENT_ERROR) {
        printf("[Port 9999] Error occurred\n");
    }
    bufferevent_free(bev);
}

// 端口9998的事件回调
static void event_cb_9998(struct bufferevent *bev, short events, void *ctx) {
    if (events & BEV_EVENT_EOF) {
        printf("[Port 9998] Connection closed\n");
    } else if (events & BEV_EVENT_ERROR) {
        printf("[Port 9998] Error occurred\n");
    }
    bufferevent_free(bev);
}
// 处理端口9999的连接事件
static void accept_conn_cb_9999(struct evconnlistener *listener, evutil_socket_t fd,
                              struct sockaddr *address, int socklen, void *ctx) {
    struct event_base *base = evconnlistener_get_base(listener);
    struct bufferevent *bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);

    if (!bev) {
        fprintf(stderr, "Error creating bufferevent for port 9999!\n");
        close(fd);
        return;
    }

    // 设置读写回调函数（使用9999端口专用的回调）
    bufferevent_setcb(bev, read_cb_9999, NULL, event_cb_9999, NULL);
    bufferevent_enable(bev, EV_READ);

    printf("New connection on port 9999 from %s:%d\n",
           inet_ntoa(((struct sockaddr_in *)address)->sin_addr),
           ntohs(((struct sockaddr_in *)address)->sin_port));
}

// 处理端口9998的连接事件
static void accept_conn_cb_9998(struct evconnlistener *listener, evutil_socket_t fd,
                              struct sockaddr *address, int socklen, void *ctx) {
    struct event_base *base = evconnlistener_get_base(listener);
    struct bufferevent *bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);

    if (!bev) {
        fprintf(stderr, "Error creating bufferevent for port 9998!\n");
        close(fd);
        return;
    }

    // 设置读写回调函数（使用9998端口专用的回调）
    bufferevent_setcb(bev, read_cb_9998, NULL, event_cb_9998, NULL);
    bufferevent_enable(bev, EV_READ);

    printf("New connection on port 9998 from %s:%d\n",
           inet_ntoa(((struct sockaddr_in *)address)->sin_addr),
           ntohs(((struct sockaddr_in *)address)->sin_port));
}

// 端口9999的数据读取回调
static void read_cb_9999(struct bufferevent *bev, void *ctx) {
    struct evbuffer *input = bufferevent_get_input(bev);
    size_t len = evbuffer_get_length(input);
    char *data = (char *)malloc(len + 1);

    if (data) {
        evbuffer_remove(input, data, len);
        data[len] = '\0';
        printf("[Port 9999] Received %zu bytes: %s\n", len, data);
        free(data);
    }
}

// 端口9998的数据读取回调
static void read_cb_9998(struct bufferevent *bev, void *ctx) {
    struct evbuffer *input = bufferevent_get_input(bev);
    size_t len = evbuffer_get_length(input);
    char *data = (char *)malloc(len + 1);

    if (data) {
        evbuffer_remove(input, data, len);
        data[len] = '\0';
        printf("[Port 9998] Received %zu bytes: %s\n", len, data);
        free(data);
    }
}



// 错误处理回调函数
static void accept_error_cb(struct evconnlistener *listener, void *ctx) {
    struct event_base *base = evconnlistener_get_base(listener);
    int err = EVUTIL_SOCKET_ERROR();
    fprintf(stderr, "Listener error %d (%s). Shutting down.\n",
            err, evutil_socket_error_to_string(err));
    event_base_loopexit(base, NULL);
}

int main(int argc, char **argv) {

    sockets::dv_qkd_tcp_server dv_qkd_tcp_server;
    dv_qkd_tcp_server.start();

    struct event_base *base;
    struct evconnlistener *listener1, *listener2;
    struct sockaddr_in sin1, sin2;

    base = event_base_new();
    if (!base) {
        fprintf(stderr, "Could not initialize libevent!\n");
        return 1;
    }

    // 设置并启动第一个监听器（端口9999）
    memset(&sin1, 0, sizeof(sin1));
    sin1.sin_family = AF_INET;
    sin1.sin_addr.s_addr = htonl(INADDR_ANY);
    sin1.sin_port = htons(9999);

    listener1 = evconnlistener_new_bind(base, accept_conn_cb_9999, NULL,
                                        LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE, -1,
                                        (struct sockaddr *)&sin1, sizeof(sin1));
    if (!listener1) {
        fprintf(stderr, "Could not create listener1!\n");
        return 1;
    }
    evconnlistener_set_error_cb(listener1, accept_error_cb);

    // 设置并启动第二个监听器（端口9998）
    memset(&sin2, 0, sizeof(sin2));
    sin2.sin_family = AF_INET;
    sin2.sin_addr.s_addr = htonl(INADDR_ANY);
    sin2.sin_port = htons(9998);

    listener2 = evconnlistener_new_bind(base, accept_conn_cb_9998, NULL,
                                        LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE, -1,
                                        (struct sockaddr *)&sin2, sizeof(sin2));
    if (!listener2) {
        fprintf(stderr, "Could not create listener2!\n");
        return 1;
    }
    evconnlistener_set_error_cb(listener2, accept_error_cb);

    printf("Server started. Listening on ports 9999 and 9998...\n");

    event_base_dispatch(base);

    evconnlistener_free(listener1);
    evconnlistener_free(listener2);
    event_base_free(base);

    return 0;
}